/**
 *
 *  Copyright (c) 2011, AstroDAbis
 *  All rights reserved.
 *
 *  Redistribution and use in source and binary forms, with or without modification,
 *  are permitted provided that the following conditions are met:
 *
 *      * Redistributions of source code must retain the above copyright notice,
 *        this list of conditions and the following disclaimer.
 *      * Redistributions in binary form must reproduce the above copyright notice,
 *        this list of conditions and the following disclaimer in the documentation
 *        and/or other materials provided with the distribution.
 *
 *  THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND
 *  ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
 *  WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED.
 *  IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT,
 *  INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING,
 *  BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
 *  DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF
 *  LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE
 *  OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED
 *  OF THE POSSIBILITY OF SUCH DAMAGE.
 *
 */
package org.jiscinvolve.astrodabis.mallow.impl.model.jpa;

import java.util.Date;
import java.util.List;
import java.util.Iterator;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

//
// Spring HibernateTemplate.
import org.springframework.orm.hibernate3.HibernateTemplate; 
import org.springframework.orm.hibernate3.support.HibernateDaoSupport; 
//
// Spring autowire annotations.
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
//
// Spring stereotype and context.
import org.springframework.stereotype.Component;
import org.springframework.stereotype.Repository;
import org.springframework.context.ApplicationContext;
//
// Spring transactions
import org.springframework.transaction.annotation.Propagation;
import org.springframework.transaction.annotation.Transactional; 
//
// Spring thread pool
import org.springframework.core.task.TaskExecutor;

//
// Bean interfaces.
import org.jiscinvolve.astrodabis.mallow.model.api.Tag;
import org.jiscinvolve.astrodabis.mallow.model.api.TagBean;
import org.jiscinvolve.astrodabis.mallow.model.api.Query;
import org.jiscinvolve.astrodabis.mallow.model.api.QueryBean;
import org.jiscinvolve.astrodabis.mallow.model.api.QueryTagBean;
import org.jiscinvolve.astrodabis.mallow.model.api.Result;
import org.jiscinvolve.astrodabis.mallow.model.api.Account;
import org.jiscinvolve.astrodabis.mallow.model.api.AccountManager;

import org.jiscinvolve.astrodabis.mallow.model.api.JoinData;
import org.jiscinvolve.astrodabis.mallow.model.api.JoinTag;
import org.jiscinvolve.astrodabis.mallow.model.api.JoinTagData;

import org.jiscinvolve.astrodabis.mallow.impl.model.api.JoinTagImpl;

import org.jiscinvolve.astrodabis.mallow.model.tap.TapJoin;

import org.jiscinvolve.astrodabis.mallow.model.tap.TapTable;
import org.jiscinvolve.astrodabis.mallow.model.tap.TapTableBean;

import org.jiscinvolve.astrodabis.mallow.model.tap.TapService;
import org.jiscinvolve.astrodabis.mallow.model.tap.TapServiceBean;

import org.jiscinvolve.astrodabis.mallow.model.jpa.PersistentObject;
import org.jiscinvolve.astrodabis.mallow.model.jpa.PersistentManager;
import org.jiscinvolve.astrodabis.mallow.model.jpa.PersistentCollection;
import org.jiscinvolve.astrodabis.mallow.model.jpa.DuplicateEntryException;

import org.jiscinvolve.astrodabis.mallow.impl.model.api.TagImpl;
import org.jiscinvolve.astrodabis.mallow.impl.model.api.QueryImpl;
import org.jiscinvolve.astrodabis.mallow.impl.model.api.QueryTagImpl;
import org.jiscinvolve.astrodabis.mallow.impl.model.api.ResultImpl;
import org.jiscinvolve.astrodabis.mallow.impl.model.api.AccountImpl;

import org.jiscinvolve.astrodabis.mallow.impl.model.tap.TapJoinImpl;
import org.jiscinvolve.astrodabis.mallow.impl.model.tap.TapTableImpl;
import org.jiscinvolve.astrodabis.mallow.impl.model.tap.TapServiceImpl;

import org.jiscinvolve.astrodabis.mallow.util.GenericIterable;

/**
 * PersistentObject manager implemetation.
 *
 */
@Repository
public class PersistentManagerImpl
extends HibernateDaoSupport
implements PersistentManagerExtra
    {

    /**
     * Our debug logger.
     * 
     */
    private static Log log = LogFactory.getLog(PersistentManagerImpl.class);

    /**
     * Get our Spring HibernateTemplate.
     * 
     */
    public HibernateTemplate template()
        {
        return this.getHibernateTemplate();
        }

    /**
     * Set our Spring HibernateTemplate.
     * Although this is inherited from JpaDaoSupport, it needs to be defined in this class
     * for the @Autowired annotation to be picked up in the package scan.
     * 
     */
    @Autowired
    public void setTemplate(HibernateTemplate template)
        {
        log.debug("PersistentManagerImpl.setTemplate(HibernateTemplate)");
        super.setHibernateTemplate(
            template
            );
        }

    /**
     * Our Spring ApplicationContext.
     * 
     */
    private static ApplicationContext context;

    /**
     * Get our Spring ApplicationContext.
     * 
     */
    public ApplicationContext context()
        {
        return this.context ;
        }

    /**
     * Set our Spring ApplicationContext.
     * 
     */
    @Autowired
    public void setContext(ApplicationContext context)
        {
        log.debug("PersistentManagerImpl.setContext(ApplicationContext)");
        if (this.context == null)
            {
            this.context = context ;
            }
        else {
            log.error("Modifying existing PersistentManagerImpl.ApplicationContext");
            }
        }

    /**
     * Our singleton instance.
     * We need a static (annotated) instance that PersistentObjects can use.
     * 
     */
    private static PersistentManagerExtra instance ;

    /**
     * Get our singleton instance.
     * This delays the wiring of the PersistentManagerExtra instance
     * until after the rest of the beans have been created, avoiding
     * circular dependencies.
     * 
     */
    public static PersistentManagerExtra instance()
        {
        log.debug("PersistentManagerImpl.instance()");
        if (instance == null)
            {
            log.debug("Static instance is null, attempting to create a new one");
            instance = (PersistentManagerExtra) context.getBean(
                "persistentManagerImpl"
                );
            }
        return instance;
        }

    /**
     * Set our singleton instance.
     * Autowiring this causes a circular dependency, and Spring context initialization fails.
     * 
    @Autowired
    @Qualifier("persistentManagerImpl")
    public void setInstance(PersistentManagerExtra instance)
        {
        log.debug("FROG PersistentManagerImpl.instance()");
        this.instance = instance ;
        }
     */

    /**
     * Protected constructor..
     * 
     */
    protected PersistentManagerImpl()
        {
        log.debug("PersistentManagerImpl()");
/*
 * This doesn't work, because we need access to the annotated proxy
 * rather than the native object.
 *
        if (null == singleton)
            {
            singleton = this ;
            }
        else {
            log.error("PersistentManagerImpl instance already assigned [" + singleton + "]");
            }
 */
        }

    @Override
    @Transactional(readOnly=false)
    public void save(PersistentObjectImpl object)
        {
        log.debug("PersistentManagerImpl.save(PersistentObjectImpl)");
        this.template().saveOrUpdate(
            object
            );  
        }

    @Override
    @Transactional(readOnly=true)
    public void refresh(PersistentObjectImpl object)
        {
        log.debug("PersistentManagerImpl.update(PersistentObjectImpl)");
        this.template().refresh(
            object
            );  
        }

    @Override
    @Transactional(readOnly=false)
    public void delete(PersistentObjectImpl object)
        {
        log.debug("PersistentManagerImpl.delete(PersistentObjectImpl)");
//TODO
        }

// Accounts 

    @Override
    public AccountManager accounts()
        {
        return new AccountManager()
            {
// This doesn't create a new (write) transaction.
            @Override
            @Transactional(readOnly=false)
            public AccountImpl create(String name, String mail, String pass)
            throws DuplicateEntryException
                {
                log.debug("PersistentManagerImpl.AccountManager.create(String, String, String)");
                return register(
                    name,
                    mail,
                    pass
                    );
                }

            @Override
            @Transactional(readOnly = true)
            public AccountImpl select(String ident)
                {
                return template().get(
                    AccountImpl.class,
                    ident
                    );  
                }

            @Override
            @Transactional(readOnly = true)
            public Iterable<Account> select()
                {
                return template().find(
                    "FROM AccountImpl ORDER BY name ASC"
                    );  
                }

            @Override
            @Transactional(readOnly = true)
            public AccountImpl login(String name, String pass)
                {
                log.debug("PersistentManagerImpl.AccountManager.login(String, String)");
                log.debug("  Name [" + name + "]");
                log.debug("  Pass [" + pass + "]");
                List<AccountImpl> list = template().find(
                    "FROM AccountImpl WHERE name= ? AND pass=?",
                    name,
                    pass
                    );  
                if (list.isEmpty())
                    {
                    return null ;
                    }
                else {
                    return list.get(0);
                    }
                }

            @Override
            @Transactional(readOnly = true)
            public AccountImpl name(String name)
                {
                log.debug("PersistentManagerImpl.AccountManager.name(String)");
                log.debug("  Name [" + name + "]");
                List<AccountImpl> list = template().find(
                    "FROM AccountImpl WHERE name= ?",
                    name
                    );  
                if (list.isEmpty())
                    {
                    return null ;
                    }
                else {
                    return list.get(0);
                    }
                }
            };
        }

    /**
     * Register a new account.
     *
     */
    @Override
    @Transactional(readOnly=false)
    public AccountImpl register(String name, String mail, String pass)
    throws DuplicateEntryException
        {
        log.debug("PersistentManagerImpl.register(String, String, String)");
        //
        // Check for duplicate name.
        List<AccountImpl> matchname = template().find(
            "FROM AccountImpl WHERE name= ?",
            name
            );  
        if (matchname.isEmpty() == false)
            {
            throw new DuplicateEntryException(
                "name"
                );
            }
        //
        // Check for duplicate email.
        List<AccountImpl> matchmail = template().find(
            "FROM AccountImpl WHERE email= ?",
            mail
            );  
        if (matchmail.isEmpty() == false)
            {
            throw new DuplicateEntryException(
                "email"
                );
            }
        //
        // Create the new account.
        AccountImpl account =  new AccountImpl(
            name,
            mail,
            pass
            );
        template().save(
            account
            );  
        return account ;
        }



// Tags

    @Override
    @Transactional(readOnly=false)
    public TagImpl tag(AccountImpl owner, TagBean bean)
        {
        log.debug("PersistentManagerImpl.tag(AccountImpl, TagBean)");
        TagImpl tag = new TagImpl(
            owner,
            bean
            );
        template().saveOrUpdate(
            tag
            );  
        return tag ;
        }

    @Override
    @Transactional(readOnly = true)
    public Iterable<Tag> tags(AccountImpl owner)
        {
        return new GenericIterable<Tag, TagImpl>(
            template().find(
                "FROM TagImpl WHERE owner = ? ORDER BY created DESC",
                owner
                )
            );  
        }

    @Override
    @Transactional(readOnly = false)
    public TagImpl tag(AccountImpl owner, String name)
        {
        log.debug("PersistentManagerImpl.tag(AccountImpl, String name)");
        TagImpl tag = new TagImpl(
            owner,
            name
            );
        template().saveOrUpdate(
            tag
            );  
        return tag ;
        }

    @Override
    public PersistentCollection<Tag, String> tags()
        {
        return new PersistentCollection<Tag, String>()
            {
            @Override
            @Transactional(readOnly = true)
            public Tag select(String ident)
                {
                return template().get(
                    TagImpl.class,
                    ident
                    );  
                }

            @Override
            @Transactional(readOnly = true)
            public Iterable<Tag> select()
                {
                return new GenericIterable<Tag, TagImpl>(
                    template().find(
                        "FROM TagImpl ORDER BY created DESC"
                        )
                    );
                }
            };
        }

// Queries

    /*
     * Get the implementation for a Query interface.
     * @todo NotFoundException.
     * @todo Move to QueryManager implementation.
     */
    private QueryImpl instance(Query query)
        {
        if (query instanceof QueryImpl)
            {
            return (QueryImpl) query ;
            }
        else {
            return query(
                query.ident()
                );
            }
        }

    @Override
    public QueryImpl query(String ident)
        {
        return template().get(
            QueryImpl.class,
            ident
            );  
        }

    @Override
    public PersistentCollection<Query, String> queries()
        {
        return new PersistentCollection<Query, String>()
            {
            @Override
            @Transactional(readOnly = true)
            public Query select(String ident)
                {
                return query(
                    ident
                    );  
                }

            @Override
            @Transactional(readOnly = true)
            public Iterable<Query> select()
                {
                return template().find(
                    "FROM QueryImpl ORDER BY created DESC"
                    );  
                }
            };
        }

// TODO Move into implementation interface
    @Override
    @Transactional(readOnly = true)
    public Iterable<Query> queries(AccountImpl owner)
        {
        return new GenericIterable<Query, QueryImpl>(
            template().find(
                "FROM QueryImpl WHERE owner = ? ORDER BY created DESC",
                owner
                )
            );  
        }

// TODO Move into implementation interface
    @Override
    @Transactional(readOnly = false)
    public QueryImpl query(AccountImpl owner, Query.Action action, QueryBean bean)
        {
        log.debug("PersistentManagerImpl.query(AccountImpl, Query.Action, QueryBean)");
        log.debug("  Owner  [" + owner.ident() + "]");
        log.debug("  Action [" + action + "]");

// Check the action makes sense.

TapTableImpl table = table(
    bean.getTable()
    );
// Check for null.

        //
        // Create the new object and save it.
        QueryImpl query = new QueryImpl(
            owner,
            table,
            bean
            );
        template().saveOrUpdate(
            query
            );  
        //
        // If the query is ready, run it.
        if (action == Query.Action.RUN)
            {
            run(query);
            }
        return query ;
        }

// QueryExecutor

    /**
     * Our Spring TaskExecutor.
     *
     */
    private static TaskExecutor executor ;

    /**
     * Our Spring TaskExecutor.
     *
     */
    public TaskExecutor executor()
        {
        return this.executor ;
        }

    /**
     * Autowire our TaskExecutor.
     *
     */
    @Autowired
    public void setExecutor(TaskExecutor executor)
        {
        log.debug("PersistentManagerImpl.setExecutor(TaskExecutor)");
        if (this.executor == null)
            {
            this.executor = executor ;
            }
        else {
            log.error("Modifying existing PersistentManagerImpl.TaskExecutor");
            }
        }

    @Override
    @Transactional(readOnly=false)
    public void running(QueryImpl query)
        {
        log.debug("PersistentManagerImpl.running(QueryImpl query)");
        log.debug("  Ident [" + query.ident() + "]");

        this.refresh(
            query
            );

        if (query.status() != Query.Status.PENDING)
            {
            throw new IllegalStateException(
                "Can't run a query with status [" + query.status() + "]"
                );
            }
        else {

            log.debug("Setting status ..");
            query.status(
                Query.Status.RUNNING
                );
            log.debug("Saving query ..");
            this.save(
                query
                );  

            }
        }

    @Override
    @Transactional(readOnly=false)
    public void completed(QueryImpl query)
        {
        log.debug("PersistentManagerImpl.completed(QueryImpl query)");
        log.debug("  Ident [" + query.ident() + "]");

        this.refresh(
            query
            );

        if (query.status() != Query.Status.RUNNING)
            {
            throw new IllegalStateException(
                "Can't complete a query with status [" + query.status() + "]"
                );
            }
        else {

            log.debug("Setting status ..");
            query.status(
                Query.Status.COMPLETED
                );
            log.debug("Saving query ..");
            this.save(
                query
                );  

            }
        }

    @Override
    @Transactional(readOnly=false)
    public void cancelled(QueryImpl query)
        {
        log.debug("PersistentManagerImpl.cancelled(QueryImpl query)");
        log.debug("  Ident [" + query.ident() + "]");

        this.refresh(
            query
            );

        if (query.status() != Query.Status.RUNNING)
            {
            throw new IllegalStateException(
                "Can't cancel a query with status [" + query.status() + "]"
                );
            }
        else {

            log.debug("Setting status ..");
            query.status(
                Query.Status.CANCELLED
                );
            log.debug("Saving query ..");
            this.save(
                query
                );  

            }
        }

    @Override
    @Transactional(readOnly=false)
    public void failed(QueryImpl query)
        {
        log.debug("PersistentManagerImpl.failed(QueryImpl query)");
        log.debug("  Ident [" + query.ident() + "]");

        this.refresh(
            query
            );

        if (query.status() != Query.Status.RUNNING)
            {
            throw new IllegalStateException(
                "Can't fail a query with status [" + query.status() + "]"
                );
            }
        else {

            log.debug("Setting status ..");
            query.status(
                Query.Status.FAILED
                );
            log.debug("Saving query ..");
            this.save(
                query
                );  

            }
        }

    @Override
    @Transactional(readOnly=false)
    public void run(QueryImpl query)
        {
        log.debug("PersistentManagerImpl.run(QueryImpl)");
        log.debug("  Ident [" + query.ident() + "]");
        //
        // Check the current status.
        if (query.status() != Query.Status.EDITING)
            {
            throw new IllegalStateException(
                "Can't run a query with status [" + query.status() + "]"
                );
            }
        //
        // Create QueryRunner for our query.
        QueryRunner runner = new QueryRunner(
            query
            );
        //
        // Synchronize on the runner to prevent the new
        // thread from starting anything until we are done. 
        synchronized(runner)
            {
            //
            // Run the query in a separate thread.
            executor().execute(
                runner
                );
            //
            // Set the status to PENDING.
            query.status(
                Query.Status.PENDING
                );
            this.save(
                query
                );  
            }
        }

    /**
     * Inner class for executing queries.
     *
     */
    public static class QueryRunner
    implements Runnable
        {

        public QueryRunner(QueryImpl query)
            {
            this.query = query ;
            }

        private QueryImpl query ;

        public void run()
            {
            log.debug("QueryRunner.run()");
            log.debug("  Ident [" + query.ident() + "]");
            //
            // Wait for the parent thread to finish.
            synchronized(this)
                {
                query.frog();
                }
            log.debug("QueryRunner.done");
            }
        }

    @Override
    @Transactional(readOnly=true)
    public Iterable<ResultImpl> results(QueryImpl owner)
        {
        return template().find(
            "FROM ResultImpl WHERE query = ? ORDER BY row ASC",
            owner
            );  
        }

    /**
     * Create a new QueryTag for a Query.
     *
     */
    @Transactional(readOnly=false)
    public QueryTagImpl tag(AccountImpl actor, Query query, QueryTagBean bean)
        {

// Check for existing query tag, throw duplicate exception ?

        //
        // Find the matching tag (create if not found).
        TagImpl tag = actor.tags().name(
            bean.getName(),
            true
            );
        //
        // Create the QueryTag.
        QueryTagImpl qtag = new QueryTagImpl(
            actor,
            instance(
                query
                ),
            tag
            );
        template().saveOrUpdate(
            qtag
            );  
        return qtag ;
        }

// Results 

    @Override
    public PersistentCollection<Result, Long> results()
        {
        return new PersistentCollection<Result, Long>()
            {
            @Override
            @Transactional(readOnly = true)
            public Result select(Long ident)
                {
                return template().get(
                    ResultImpl.class,
                    ident
                    );  
                }

            @Override
            @Transactional(readOnly = true)
            public Iterable<Result> select()
                {
                return template().find(
                    "FROM ResultImpl ORDER BY ident DESC"
                    );  
                }
            };
        }

// Services

    @Override
    public TapServiceImpl service(String ident)
        {
        return template().get(
            TapServiceImpl.class,
            ident
            );  
        }

    @Override
    public PersistentCollection<TapService, String> services()
        {
        return new PersistentCollection<TapService, String>()
            {
            @Override
            @Transactional(readOnly = true)
            public TapService select(String ident)
                {
                return service(
                    ident
                    );  
                }

            @Override
            @Transactional(readOnly = true)
            public Iterable<TapService> select()
                {
                return template().find(
                    "FROM TapServiceImpl ORDER BY created DESC"
                    );  
                }
            };
        }

// TODO Move into collection ex interface
    @Override
    @Transactional(readOnly = false)
    public TapServiceImpl service(AccountImpl owner, TapService.Action action, TapServiceBean bean)
        {
        log.debug("PersistentManagerImpl.service(AccountImpl, TapService.Action, TapServiceBean)");
        log.debug("  Owner  [" + owner.ident() + "]");
        log.debug("  Action [" + action + "]");

// Check the action makes sense.
        //
        // Create the new object and save it.
        TapServiceImpl service = new TapServiceImpl(
            owner,
            bean
            );
        template().saveOrUpdate(
            service
            );  
        return service ;
        }

// TODO Move into collection ex interface
    @Override
    @Transactional(readOnly = true)
    public Iterable<TapService> services(AccountImpl owner)
        {
        return new GenericIterable<TapService, TapServiceImpl>(
            template().find(
                "FROM TapServiceImpl WHERE owner = ? ORDER BY created DESC",
                owner
                )
            );  
        }

// Tables

    @Override
    @Transactional(readOnly = true)
    public TapTableImpl table(String ident)
        {
        return template().get(
            TapTableImpl.class,
            ident
            );  
        }

    @Override
    public PersistentCollection<TapTable, String> tables()
        {
        return new PersistentCollection<TapTable, String>()
            {
            @Override
            public TapTable select(String ident)
                {
                return table(
                    ident
                    );  
                }

            @Override
            @Transactional(readOnly = true)
            public Iterable<TapTable> select()
                {
                return template().find(
                    "FROM TapTableImpl ORDER BY created DESC"
                    );  
                }
            };
        }


// TODO Move into collection ex interface
    @Override
    @Transactional(readOnly = false)
    public TapTableImpl table(TapServiceImpl service, TapTable.Action action, TapTableBean bean)
        {
        log.debug("PersistentManagerImpl.table(TapServiceImpl, TapTable.Action, TapTableBean)");
        log.debug("  Service [" + service.ident() + "]");
        log.debug("  Action  [" + action + "]");

// Check the action makes sense.
// Check the service is valid.
        //
        // Create the new object and save it.
        TapTableImpl table = new TapTableImpl(
            service,
            bean
            );
        template().saveOrUpdate(
            table
            );  
        return table ;
        }

/*
// TODO Move into collection ex interface
    @Override
    @Transactional(readOnly = true)
    public TapTableImpl table(TapServiceImpl parent, String ident)
        {
        return template().get(
            TapTableImpl.class,
            ident
            );  
        }
 */

// TODO Move into collection ex interface
    @Override
    @Transactional(readOnly = true)
    public Iterable<TapTable> tables(TapServiceImpl parent)
        {
        return new GenericIterable<TapTable, TapTableImpl>(
            template().find(
                "FROM TapTableImpl WHERE service = ? ORDER BY created DESC",
                parent
                )
            );  
        }

    @Override
    @Transactional(readOnly = true)
    public Iterable<TapTable> tables(AccountImpl owner)
        {
        return new GenericIterable<TapTable, TapTableImpl>(
            template().find(
                "FROM TapTableImpl WHERE owner = ? ORDER BY created DESC",
                owner
                )
            );  
        }

// TapJoins

    @Override
    public JoinManagerEx joins()
        {
        return new JoinManagerEx()
            {
            @Override
            public Iterable<TapJoin> select()
                {
                return selectJoins();
                }

            @Override
            public TapJoinImpl select(String ident)
                {
                return selectJoin(
                    ident
                    );
                }

            @Override
            public TapJoinImpl select(JoinData data)
                {
                return selectJoin(
                    data
                    );
                }

            @Override
            public TapJoinImpl create(JoinData data)
                {
                return createJoin(
                    data
                    );
                }

            @Override
            public JoinTag tag(Account owner, JoinTagData data)
                {
                log.debug("PersistentManagerImpl.joins.tag(Account, JoinTagData)");
                return tagJoin(
                    owner,
                    data
                    );
                }
            };
        }

    /**
     * Find a specific TapJoin.
     *
     */
    @Transactional(readOnly = true)
    protected TapJoinImpl selectJoin(String ident)
        {
        return template().get(
            TapJoinImpl.class,
            ident
            );  
        }

    /**
     * Return an Iterable of all TapJoins (global list).
     *
     */
    @Transactional(readOnly = true)
    protected Iterable<TapJoin> selectJoins()
        {
        return template().find(
            "FROM TapJoinImpl ORDER BY created DESC"
            );  
        }

    /**
     * Find a matching join, returns null if no match found.
     *
     */
    @Transactional(readOnly = true)
    protected TapJoinImpl selectJoin(JoinData data)
        {
        List<TapJoinImpl> list = template().find(
            "FROM TapJoinImpl WHERE t1= ? AND t2=? AND p1=? AND p2=? ",
            data.t1(),
            data.t2(),
            data.p1(),
            data.p2()
            );  
        if (list.isEmpty())
            {
            return null ;
            }
        else {
            return list.get(0);
            }
        }

    /**
     * Find a matching join or create a new one.
     *
     */
    @Transactional(readOnly = false)
    protected TapJoinImpl createJoin(JoinData data)
        {
        //
        // Check for existing join.
        TapJoinImpl join = selectJoin(
            data
            );
        //
        // Create a new join.
        if (join == null)
            {
            join = new TapJoinImpl(
                data
                );
            this.save(
                join
                );
            }
        return join ;
        }

    /**
     * Create a join and apply a tag.
     *
     */
    @Transactional(readOnly = false)
    protected JoinTag tagJoin(Account owner, JoinTagData data)
        {
        log.debug("PersistentManagerImpl.tagJoin(Account, JoinTagData)");
        JoinTagImpl jointag = new JoinTagImpl(
            (AccountImpl)owner,
            createJoin(
                data
                ),
            tag(
                (AccountImpl)owner,
                data.name()
                ),
            data.value()
            );
/*
        this.save(
            jointag
            );  
 */
        template().saveOrUpdate(
            jointag
            );  
        log.debug("Saved new JoinTag");
        return jointag ;
        }

    /**
     * Iterate an Iterator inside a database transaction.
     *
     */
    @Transactional(readOnly=false, propagation=Propagation.REQUIRES_NEW)
    public void process(Iterator iter)
        {
        log.debug("PersistentManagerImpl.process().start");
        while (iter.hasNext())
            {
            log.debug("PersistentManagerImpl.process().next");
            iter.next();
            }
        log.debug("PersistentManagerImpl.process().done");
        }

    }
